package com.xfsg.core.log.config;

import com.xfsg.core.kafka.TracingKafkaAspect;
import com.xfsg.core.log.logger.PitayaLogger;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.springframework.boot.autoconfigure.AutoConfigureAfter;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.*;

import java.util.Map;

/**
 * @author ZhangYi
 * @version 1.0.0
 * @Description:
 * @date 2022/3/28
 */
@ConditionalOnClass({KafkaTemplate.class})
@EnableConfigurationProperties({KafkaProperties.class})
@AutoConfigureAfter(PitayaLogger.class)
@Configuration(proxyBeanMethods = false)
public class KafkaLogTraceAutoConfiguration {

    private final KafkaProperties properties;

    public KafkaLogTraceAutoConfiguration(KafkaProperties properties) {
        this.properties = properties;
    }

    @Bean
    public ProducerFactory<?, ?> kafkaProducerFactory() {
        // 拿出配置
        Map<String, Object> map = this.properties.buildProducerProperties();
        // 加上interceptor
        map.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG,"com.xfsg.core.kafka.v3.TracingProducerInterceptor");
        DefaultKafkaProducerFactory<?, ?> factory = new DefaultKafkaProducerFactory<>(map);
        String transactionIdPrefix = this.properties.getProducer().getTransactionIdPrefix();
        if (transactionIdPrefix != null) {
            factory.setTransactionIdPrefix(transactionIdPrefix);
        }
        return factory;
    }

    /**
     * 这个方法不行
     * @return
     */
    @Deprecated
    public ConsumerFactory<?, ?> kafkaConsumerFactory() {
        Map<String, Object> map = this.properties.buildConsumerProperties();
        map.put(ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG,"com.xfsg.core.kafka.v3.TracingConsumerInterceptor");
        return new DefaultKafkaConsumerFactory(map);
    }

    @Bean
    public TracingKafkaAspect tracingKafkaAspect(){
        return new TracingKafkaAspect();
    }


}
